package com.niit.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/22
 * Author：Ys
 * Description:
 */
object StreamingJoin05 {

  def main(args: Array[String]): Unit = {
    //环境准备
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamingJoin05")
    // --》初始化SparkStreaming环境
    val ssc = new StreamingContext(sparkConf, Seconds(5)) //Seconds(5) 采集周期，每隔5秒获取一次数据
    ssc.sparkContext.setLogLevel("ERROR")// 设置日志级别

    //Join，将两个数据流进行合并
    val lines9999: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    val lines8848: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 8848)

    val map9999: DStream[(String, String)] = lines9999.map((_, "9999"))
    val map8848: DStream[(String, String)] = lines8848.map((_, "8848"))

    val joinDS: DStream[(String, (String, String))] = map9999.join(map8848)//只有相同的Key才可以Join连接，Key不相同是没有办法合并的

    joinDS.print()

    //--> 开启采集器
    ssc.start()
    //--> 等待采集器的关闭
    ssc.awaitTermination()

  }

}
